package day05;


import akka.japi.pf.Match;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;
import untils.OrderEvent;

/**
 * 实时对账程序
 */
public class Example4 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<OrderEvent> appStream = env
                .addSource(new SourceFunction<OrderEvent>() {
                    @Override
                    public void run(SourceContext<OrderEvent> ctx) throws Exception {
                        ctx.collectWithTimestamp(new OrderEvent("order-1", "app-zhifu", 1000L), 1000L);
                        Thread.sleep(1000L);
                        ctx.collectWithTimestamp(new OrderEvent("order-2", "app-zhifu", 2000L), 2000L);
                        Thread.sleep(1000L);
                        ctx.emitWatermark(new Watermark(7000L));
                        Thread.sleep(1000L);
                    }

                    @Override
                    public void cancel() {

                    }
                });

        DataStreamSource<OrderEvent> weixinStream = env
                .addSource(new SourceFunction<OrderEvent>() {
                    @Override
                    public void run(SourceContext<OrderEvent> ctx) throws Exception {
                        ctx.collectWithTimestamp(new OrderEvent("order-1", "app-zhifu", 1000L), 1000L);
                        Thread.sleep(1000L);
                        ctx.collectWithTimestamp(new OrderEvent("order-2", "app-zhifu", 2000L), 2000L);
                        Thread.sleep(1000L);
                        ctx.emitWatermark(new Watermark(7000L));
                        Thread.sleep(1000L);
                    }

                    @Override
                    public void cancel() {

                    }
                });
        appStream.keyBy(r -> r.orderId)
                .connect(weixinStream.keyBy(r -> r.orderId))
                .process(new Match())
                .print();

        env.execute();
    }

    public static class Match extends CoProcessFunction<OrderEvent,OrderEvent,String>{

        ValueState<OrderEvent> appstate;
        ValueState<OrderEvent> weixinstate;

        @Override
        public void open(Configuration parameters) throws Exception {
            appstate = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("app-state", Types.POJO(OrderEvent.class)));
            weixinstate = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("weixin-state", Types.POJO(OrderEvent.class)));
        }

        @Override
        public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
            if(weixinstate.value() == null){
                appstate.update(value);
                ctx.timerService().registerEventTimeTimer(value.ts + 5000L);
            } else {
                out.collect("订单" + value.orderId + "对账成功,weixin支付先到达,app支付后到达");
                weixinstate.clear();
            }
        }

        @Override
        public void processElement2(OrderEvent value, Context ctx, Collector<String> out) throws Exception {

            if(weixinstate.value() == null){
                appstate.update(value);
                ctx.timerService().registerEventTimeTimer(value.ts + 5000L);
            } else {
                out.collect("订单" + value.orderId + "对账成功,app支付先到达,weixin支付后到达");
                weixinstate.clear();
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            if (appstate != null){
                out.collect("订单" + appstate.value().orderId + "对账失败,weixin支付5秒内未到达");
                appstate.clear();
            }
            if (weixinstate != null){
                out.collect("订单" + weixinstate.value().orderId + "对账失败,weixin支付5秒内未到达");
                weixinstate.clear();
            }
        }
    }
}
